package org.jetlinks.community.messaging.rabbitmq.impl;

import org.jetlinks.community.messaging.rabbitmq.RabbitMQConsumer;
import org.jetlinks.community.messaging.rabbitmq.RabbitMQDataSource;
import org.jetlinks.community.messaging.rabbitmq.ReactorRabbitMQConsumer;
import org.jetlinks.community.messaging.rabbitmq.command.CreateConsumer;
import org.jetlinks.community.datasource.CommandHandler;
import org.jetlinks.community.datasource.DataSource;
import reactor.core.publisher.Mono;

public class CreateConsumerHandler implements CommandHandler<CreateConsumer, Mono<RabbitMQConsumer>> {

    @Override
    public Mono<RabbitMQConsumer> handle(DataSource dataSource, CreateConsumer command) {
        ReactorRabbitMQConsumer consumer = new ReactorRabbitMQConsumer(command.getExchangeOrQueue(),
                                                                       command.isTopic(),
                                                                       command.getProperties() == null
                                                                           ? dataSource
                                                                           .unwrap(RabbitMQDataSource.class)
                                                                           .getConfig()
                                                                           : command.getProperties())
            .autoAck(command.isAutoAck())
            .autoCreateTopic(command.isAutoCreate())
            .consumerGroup(command.getConsumerGroup())
            .consumerRouteKey(command.getRouteKey());


        return consumer
            .initAsync()
            .thenReturn(consumer);
    }
}
